package cn.doitedu.demo1;

import cn.doitedu.beans.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.elasticsearch.common.recycler.Recycler;

import java.util.List;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/12
 * @Desc: 学大数据，上多易教育
 *  规则：
 *    动态画像统计条件：
 *        最近 3次加购物车事件的商品平均价格 > 300
 *        最近 3次加购物车事件的最小商品价格 > 50
 *
 *    触发条件
 *       当它签到(event_id = "QD")的时候，发送营销消息
 *
 *    {"user_id":1,"event_id":"add_cart","properties":{"item_id":1,"item_price":100,"item_name":"苹果"},"action_time":1652400000000}
 *
 **/
public class Exercise {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        // 构造kafkaSource
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("ods-user-action-log")
                .setGroupId("ggg")
                .setClientIdPrefix("client-001")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();

        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source");

        SingleOutputStreamOperator<EventBean> beanStream = stream.map(json -> JSON.parseObject(json, EventBean.class));

        beanStream.keyBy(EventBean::getUser_id)
                .process(new KeyedProcessFunction<Long, EventBean, String>() {

                    ValueState<List<EventBean>> state;
                    JSONObject message ;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        state = getRuntimeContext().getState(new ValueStateDescriptor<List<EventBean>>("rule-2-2-lst", TypeInformation.of(new TypeHint<List<EventBean>>() {
                        })));


                        message = new JSONObject();
                        message.put("rule_id","rule-2-2");

                    }

                    @Override
                    public void processElement(EventBean eventBean, KeyedProcessFunction<Long, EventBean, String>.Context ctx, Collector<String> out) throws Exception {
                        String eventId = eventBean.getEvent_id();
                        long userId = eventBean.getUser_id();
                        long actionTime = eventBean.getAction_time();

                        if("QD".equals(eventId)){

                            // 判断加购事件列表中的  商品平均价格  和 最小价格
                            List<EventBean> list = state.value();
                            if(list != null && list.size()>=3 ){
                                int cnt = 0;
                                double priceSum = 0;
                                double minPrice = Double.MAX_VALUE;
                                for (EventBean bean : list) {
                                    double price = Double.parseDouble(bean.getProperties().get("item_price"));
                                    priceSum += price;
                                    cnt++;
                                    if(price < minPrice){
                                        minPrice = price;
                                    }
                                }

                                if(priceSum/cnt > 300 && minPrice > 50){
                                    message.put("user_id",userId);
                                    message.put("hit_time",actionTime);
                                    out.collect(message.toJSONString());
                                }


                            }



                        }


                        List<EventBean> list = state.value();
                        if("add_cart".equals(eventId)){
                            if(list== null){
                                state.update(Lists.newArrayList(eventBean));
                            }else if(list.size()<3){
                                list.add(eventBean);
                            }else{
                                list.remove(0);
                                list.add(eventBean);

                            }

                        }
                    }
                });





    }

}
